package com.rjhch.bigDataCluterHelp.hbaseHelp;

import com.rjhch.bigDataCluterHelp.CommonTools;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;

/**
 * 测试hbase表自动处理，包含合并region、拆分region、大合并、平衡regionServer
 */
public class HbaseTableAutoProcess extends Configured implements Tool {
    private static final Logger log = LoggerFactory.getLogger(HbaseTableAutoProcess.class);

    private static boolean regionSplitEnable;       //region拆分使能开关
    private static boolean regionMergeEnable;       //region合并使能开关
    private static boolean regionMergeForcible;     //region强制合并使能开关，当hbase.region.merge.enable为true时有效
    private static boolean tableMajorCompactEnable;     //表大合并使能开关
    private static boolean balancerEnable;              //region平衡分布使能开关

    private static int tableSplitTimes;             //拆分轮次
    private static int splitInterval;               //每一轮拆分region的时间间隔（单位：ms）
    private static long sizeLeast;                  //表和region拆分的最小阈值（单位：byte）
    private static long sizeLeastOfIndex;           //索引表和region拆分的最小阈值（单位：byte）
    private static String[] checkTables;          //namespace为default中，需要检测的表
    private static String[] noCheckTables;          //namespace为default中，不需要检测的表

    private static String[] indexes;           //索引表前缀数组

    /**
     * 初始化系统
     */
    private void init() throws Exception {
        CommonTools.init();

        regionSplitEnable = CommonTools.getBoolean("hbase.region.split.enable", false);
        regionMergeEnable = CommonTools.getBoolean("hbase.region.merge.enable", false);
        regionMergeForcible = CommonTools.getBoolean("hbase.region.merge.forcible", false);
        tableMajorCompactEnable = CommonTools.getBoolean("hbase.table.majorCompact.enable", false);
        balancerEnable = CommonTools.getBoolean("hbase.region.balance.enable", false);

        tableSplitTimes = CommonTools.getInt("hbase.table.split.times", 5);
        splitInterval = CommonTools.getInt("hbase.region.split.interval", 10);
        sizeLeast = CommonTools.getLong("hbase.check.size.least", 21474836480L);       //20G
        sizeLeastOfIndex = CommonTools.getLong("hbase.check.index.size.least", 1073741824L);       //1G
        noCheckTables = CommonTools.config.getStrings("hbase.table.no.check", new String[]{"SYSTEM.CATALOG", "SYSTEM.FUNCTION", "SYSTEM.MUTEX", "SYSTEM.SEQUENCE", "SYSTEM.STATS"});
        checkTables = CommonTools.config.getStrings("hbase.table.check", new String[]{});

        indexes = CommonTools.config.getStrings("hbase.table.indexes", new String[]{"index"});

        log.info("regionSplitEnable: " + regionSplitEnable);
        log.info("regionMergeEnable: " + regionMergeEnable);
        log.info("regionMergeForcible: " + regionMergeForcible);
        log.info("tableMajorCompactEnable: " + tableMajorCompactEnable);
        log.info("balancerEnable: " + balancerEnable);

        log.info("tableSplitTimes: " + tableSplitTimes);
        log.info("splitInterval: " + splitInterval);
        log.info("sizeLeast: " + sizeLeast);
        log.info("sizeLeastOfIndex: " + sizeLeastOfIndex);
        log.info("checkTables: " + Arrays.asList(checkTables));
        log.info("noCheckTables: " + Arrays.asList(noCheckTables));
        log.info("indexes: " + Arrays.asList(indexes));
    }

    /**
     * 是否需要检测的表
     *
     * @param tableName
     * @return
     */
    private boolean isNeedCheckTableBase(TableName tableName) throws Exception {
        log.debug("check table: " + tableName.getNameAsString());
        if (!"default".equals(tableName.getNamespaceAsString())) {        //default命名空间的才需要检测
            log.debug("table's namespace is not default. namespace is: " + tableName.getNamespaceAsString());
            return false;
        }

        if (checkTables.length != 0 ){
            for (String check_table : checkTables) {             //需要检测的表，直接返回true
                if (tableName.getNameAsString().matches(check_table)) {
                    log.debug("table is in check tables. ");
                    return true;
                }
            }
            log.debug("check table is not empty and this table is not in check tables. ");
            return false;   //需要检测的列表中没有找到，直接返回false
        }

        for (String no_check_table : noCheckTables) {             //一些表不需要检测，节约时间
            if (tableName.getNameAsString().matches(no_check_table)) {
                log.debug("table is in no check tables. ");
                return false;
            }
        }

        return true;
    }

    /**
     * 是否需要检测的表
     *
     * @param tableName
     * @return
     */
    private boolean isNeedCheckTable(TableName tableName) throws Exception {
        if (!isNeedCheckTableBase(tableName)) {        //default命名空间的才需要检测
            return false;
        }

        Path path_table = new Path("/hbase/data/default/" + tableName.getNameAsString());
        long length_path_table = CommonTools.hdfs.getContentSummary(path_table).getLength();
        if (length_path_table < getTableOrRegionSizeLeast(tableName)) {           //大小超过1G的表才需要检测
            log.debug("table size is lease then threshold. table size: " + length_path_table + ", threshold: " + getTableOrRegionSizeLeast(tableName));
            return false;
        }

        return true;
    }

    /**
     * 是否需要检测的region
     *
     * @param regionInfo
     * @return
     */
    private boolean isNeedCheckRegion(TableName tableName, HRegionInfo regionInfo) throws Exception {
        long length_path_region = getRegionSize(tableName, regionInfo);
        if (length_path_region < getTableOrRegionSizeLeast(tableName)) {           //大小超过指定大小的表才需要检测
            log.debug("region size is lease then threshold. region size: " + length_path_region + ", threshold: " + getTableOrRegionSizeLeast(tableName));
            return false;
        }
        return true;
    }

    /**
     * 获取表中处于在线状态的regions
     *
     * @param tableName
     * @return
     */
    private List<HRegionInfo> getOnlineRegions(TableName tableName) throws Exception {
        List<HRegionInfo> retRegionInfos = new ArrayList<>();
        List<HRegionInfo> hRegionInfos = CommonTools.admin.getTableRegions(tableName);
        for (HRegionInfo hRegionInfo : hRegionInfos) {
            if (!hRegionInfo.isOffline()) {
                retRegionInfos.add(hRegionInfo);
            }
        }
        log.debug("online region size in table: " + tableName + " is " + retRegionInfos.size());
        return retRegionInfos;
    }

    /**
     * 获取region大小
     *
     * @param regionInfo
     * @return
     */
    private long getRegionSize(TableName tableName, HRegionInfo regionInfo) throws Exception {
        Path path_region = new Path("/hbase/data/default/" + tableName.getNameAsString() + "/" + regionInfo.getShortNameToLog());
        long length_path_region = CommonTools.hdfs.getContentSummary(path_region).getLength();
        log.debug("region: " + regionInfo.getRegionNameAsString() + ", region size:" + length_path_region);
        return length_path_region;
    }

    /**
     * 获取region大小map
     *
     * @param hRegionInfos
     * @return
     */
    private Map<HRegionInfo, Long> getRegionSizes(TableName tableName, List<HRegionInfo> hRegionInfos) throws Exception {
        Map<HRegionInfo, Long> hRegionSizes = new HashMap<>();
        for (HRegionInfo hRegionInfo : hRegionInfos) {
            hRegionSizes.put(hRegionInfo, getRegionSize(tableName, hRegionInfo));
        }
        return hRegionSizes;
    }

    /**
     * 是否是索引表
     *
     * @param tableName
     * @return
     */
    private boolean isIndexTable(TableName tableName) {
        for (String index : indexes) {
            if (tableName.getNameAsString().matches(index)) {
                log.debug("table {} is an index table.", tableName.getNameAsString());
                return true;
            }
        }
        return false;
    }

    /**
     * 获取表和region对应的检测大小
     *
     * @param tableName
     * @return
     */
    private long getTableOrRegionSizeLeast(TableName tableName) {
        return isIndexTable(tableName) ? sizeLeastOfIndex : sizeLeast;
    }

    /**
     * 获取可以merge的RegionInfo对的列表（不强制，只会合并相邻region）
     *
     * @param regionSizeMap
     * @return
     */
    private List<List<HRegionInfo>> getNoForcibleMergeRegionInfos(Map<HRegionInfo, Long> regionSizeMap) {
        List<List<HRegionInfo>> mergeRegionInfos = new ArrayList<>();
        if (regionSizeMap.size() < 2) {
            return mergeRegionInfos;
        }

        //获取大小和不超过sizeLeast的两个相邻region
        List<HRegionInfo> tmpRegionInfos = new ArrayList<>();
        boolean isFind = false;

        for (Map.Entry<HRegionInfo, Long> regionSizeEntry : regionSizeMap.entrySet()) {
            for (Map.Entry<HRegionInfo, Long> regionSizeEntry2 : regionSizeMap.entrySet()) {
                String endKeyOfregionSizeEntry = new String(regionSizeEntry.getKey().getEndKey());
                String startKeyOfregionSizeEntry2 = new String(regionSizeEntry2.getKey().getStartKey());
                if (!"".equals(endKeyOfregionSizeEntry) && endKeyOfregionSizeEntry.equals(startKeyOfregionSizeEntry2)) {   //一个region的endKey和另一个的startKey一样时，才可以合并
                    if (regionSizeEntry.getValue() + regionSizeEntry2.getValue() < getTableOrRegionSizeLeast(regionSizeEntry.getKey().getTable())) { //如果两个region的大小之和小于指定阈值，则可以进行合并
                        tmpRegionInfos.add(regionSizeEntry2.getKey());
                        tmpRegionInfos.add(regionSizeEntry.getKey());
                        isFind = true;
                        break;
                    }
                }
            }
            if (isFind) {
                break;
            }
        }

        if (isFind) {
            mergeRegionInfos.add(tmpRegionInfos);

            //从regionSizeMap中去除获取到的这两个region，再去获取
            regionSizeMap.remove(tmpRegionInfos.get(0));
            regionSizeMap.remove(tmpRegionInfos.get(1));

            List<List<HRegionInfo>> mergeRegionInfos2 = getNoForcibleMergeRegionInfos(regionSizeMap);
            if (!mergeRegionInfos2.isEmpty()) {
                mergeRegionInfos.addAll(mergeRegionInfos2);
            }
        }

        return mergeRegionInfos;
    }

    /**
     * 获取可以merge的RegionInfo对的列表
     *
     * @param regionSizeMap
     * @return
     */
    private List<List<HRegionInfo>> getForcibleMergeRegionInfosInner(Map<HRegionInfo, Long> regionSizeMap) {
        List<List<HRegionInfo>> mergeRegionInfos = new ArrayList<>();
        if (regionSizeMap.size() < 2) {
            return mergeRegionInfos;
        }

        //获取大小和不超过sizeLeast的两个region
        List<HRegionInfo> tmpRegionInfos = new ArrayList<>();
        boolean isFind = false;

        for (Map.Entry<HRegionInfo, Long> regionSizeEntry : regionSizeMap.entrySet()) {
            for (Map.Entry<HRegionInfo, Long> regionSizeEntry2 : regionSizeMap.entrySet()) {
                if (regionSizeEntry.getKey().equals(regionSizeEntry2.getKey())) {
                    continue;
                }

                if (regionSizeEntry.getValue() + regionSizeEntry2.getValue() < getTableOrRegionSizeLeast(regionSizeEntry.getKey().getTable())) { //如果两个region的大小之和小于指定阈值，则可以进行合并
                    tmpRegionInfos.add(regionSizeEntry.getKey());
                    tmpRegionInfos.add(regionSizeEntry2.getKey());
                    isFind = true;
                    break;
                }
            }
            if (isFind) {
                break;
            }
        }

        if (isFind) {
            mergeRegionInfos.add(tmpRegionInfos);

            //从regionSizeMap中去除获取到的这两个region，再去获取
            regionSizeMap.remove(tmpRegionInfos.get(0));
            regionSizeMap.remove(tmpRegionInfos.get(1));

            List<List<HRegionInfo>> mergeRegionInfos2 = getForcibleMergeRegionInfosInner(regionSizeMap);
            if (!mergeRegionInfos2.isEmpty()) {
                mergeRegionInfos.addAll(mergeRegionInfos2);
            }
        }

        return mergeRegionInfos;
    }

    /**
     * 获取可以merge的RegionInfo对的列表（强制的，不相邻region也可以合并）
     *
     * @param regionSizeMap
     * @return
     */
    private List<List<HRegionInfo>> getForcibleMergeRegionInfos(Map<HRegionInfo, Long> regionSizeMap) {
        List<List<HRegionInfo>> mergeRegionInfos = new ArrayList<>();
        List<List<HRegionInfo>> noForcibleMergeRegionInfos = getNoForcibleMergeRegionInfos(regionSizeMap);      //先取能连续合并的
        log.debug("The number of region pairs that can be combined continuously is " + noForcibleMergeRegionInfos.size());
        mergeRegionInfos.addAll(noForcibleMergeRegionInfos);      //先取能连续合并的

        List<List<HRegionInfo>> forcibleMergeRegionInfos = getForcibleMergeRegionInfosInner(regionSizeMap);      //再取不能连续合并的，regionSizeMap中数量在getMergeRegionInfos中会减少，这里不需要去除
        log.debug("The number of region pairs that will be forced to merge after continuous merge is " + forcibleMergeRegionInfos.size());

        mergeRegionInfos.addAll(forcibleMergeRegionInfos);      //再取不能连续合并的，regionSizeMap中数量在getMergeRegionInfos中会减少，这里不需要去除
        return mergeRegionInfos;
    }

    /**
     * 自动合并region
     *
     * @throws Exception
     */
    private void autoMergeRegion() throws Exception {
        TableName[] tableNames = CommonTools.admin.listTableNames();
        for (TableName tableName : tableNames) {
            isNeedCheckTableBase(tableName);

            boolean isRegionMerge = false;    //是否有reigon在进行合并
            List<HRegionInfo> hRegionInfos = getOnlineRegions(tableName);
            if (hRegionInfos.size() < 2) {
                log.debug("region count is lease then 2, the size: " + hRegionInfos.size());
                continue;
            }

            Map<HRegionInfo, Long> hRegionSizes = getRegionSizes(tableName, hRegionInfos);
            List<List<HRegionInfo>> mergeRegionInfos;
            if (regionMergeForcible) {
                mergeRegionInfos = getForcibleMergeRegionInfos(hRegionSizes);
                log.debug("The all number of region pairs that will be forced to merge is " + mergeRegionInfos.size());
            } else {
                mergeRegionInfos = getNoForcibleMergeRegionInfos(hRegionSizes);
                log.debug("The number of region pairs that can be combined continuously is " + mergeRegionInfos.size());
            }

            if (!mergeRegionInfos.isEmpty()) {
                isRegionMerge = true;
            }

            log.info("The number of region pairs that will be merged is " + mergeRegionInfos.size());
            for (List<HRegionInfo> hRegionInfoWillMerge : mergeRegionInfos) {
                try {
                    log.info("merge region: " + hRegionInfoWillMerge.get(0).getRegionNameAsString() + ", " + hRegionInfoWillMerge.get(1).getRegionNameAsString());
                    CommonTools.admin.mergeRegions(hRegionInfoWillMerge.get(0).getEncodedNameAsBytes(), hRegionInfoWillMerge.get(1).getEncodedNameAsBytes(), regionMergeForcible);
                } catch (Exception e) {
                    //记录日志，但不做处理
                    log.error("", e);
                }
            }

            if (isRegionMerge) {
                log.info("sleep " + splitInterval + " ms ...");
                Thread.sleep(splitInterval);
            }
        }
    }

    /**
     * 自动拆分region
     *
     * @throws Exception
     */
    private void autoSplitRegion() throws Exception {
        TableName[] tableNames = CommonTools.admin.listTableNames();
        for (TableName tableName : tableNames) {
            if (!isNeedCheckTable(tableName)) {
                continue;
            }

            log.info("need check table name: " + tableName);
            for (int i = 0; i < tableSplitTimes; i++) {   //每个region会检测指定次数
                boolean isRegionSplit = false;    //是否有reigon在进行拆分
                log.info("start check regions, the check times: " + (i + 1));
                List<HRegionInfo> hRegionInfos = getOnlineRegions(tableName);
                for (HRegionInfo hRegionInfo : hRegionInfos) {
                    if (!isNeedCheckRegion(tableName, hRegionInfo)) {
                        continue;
                    }

                    try {
                        isRegionSplit = true;
                        log.info("split region: " + hRegionInfo.getRegionNameAsString());
                        CommonTools.admin.splitRegion(hRegionInfo.getRegionName());
                    } catch (Exception e) {
                        //记录日志，但不做处理
                        log.error("", e);
                    }
                }

                if (isRegionSplit) {
                    log.info("sleep " + splitInterval + " ms ...");
                    Thread.sleep(splitInterval);
                }
            }
        }
    }

    /**
     * 自动执行合并
     *
     * @throws Exception
     */
    private void autoMajorCompact() throws Exception {
        TableName[] tableNames = CommonTools.admin.listTableNames();
        for (TableName tableName : tableNames) {
            if (!isNeedCheckTable(tableName)) {
                continue;
            }

            try {
                //major_compact
                log.info("majorCompact table: " + tableName);
                CommonTools.admin.majorCompact(tableName);
            } catch (Exception e) {
                //记录日志，但不做处理
                log.error("", e);
            }
        }
    }

    /**
     * 自动平衡
     *
     * @throws Exception
     */
    private void autoBalance() throws Exception {

        List<RegionEntity> regionEntities = new ArrayList<>();
        List<HRegionInfo> regionInfos = new ArrayList<>();
        List<String> regionServerNames = new ArrayList<>();

        //获取在线的regionServers
        List<ServerName> regionServers = new ArrayList<>();
        regionServers.addAll(CommonTools.admin.getClusterStatus().getServers());
        ArrayList<ServerName> deadServers2 = new ArrayList<>();
        deadServers2.addAll(CommonTools.admin.getClusterStatus().getDeadServerNames());
        regionServers.removeAll(deadServers2);

        for (ServerName serverName : regionServers) {
            regionServerNames.add(serverName.getServerName());
            List<HRegionInfo> onlineRegions = CommonTools.admin.getOnlineRegions(serverName);
            for (HRegionInfo regionInfo : onlineRegions) {
                RegionEntity regionEntity = new RegionEntity(regionInfo.getRegionNameAsString(), regionInfo.getTable().getNameAsString(), serverName.getServerName());
                regionEntities.add(regionEntity);
                regionInfos.add(regionInfo);
            }
        }

        TableBalance tableBalance = new TableBalance(regionEntities, regionServerNames.toArray(new String[]{}));
        List<RegionEntity> movedRegions = tableBalance.balance();
        for (RegionEntity regionEntity : movedRegions) {
            log.info("will moved region: {}", regionEntity);
            HRegionInfo regionInfo = getHRegionInfoWith(regionInfos,regionEntity);
            ServerName serverName = getServerNameWith(regionServers,regionEntity);

            CommonTools.admin.move(regionInfo.getEncodedNameAsBytes(),serverName.getServerName().getBytes());
        }

    }

    private HRegionInfo getHRegionInfoWith(List<HRegionInfo> onlineRegions, RegionEntity regionEntity) {
        for (HRegionInfo regionInfo:onlineRegions) {
            if (regionInfo.getRegionNameAsString().equals(regionEntity.getRegionName())){
                return regionInfo;
            }
        }
        return null;
    }

    private ServerName getServerNameWith(List<ServerName> servers, RegionEntity regionEntity) {
        for (ServerName serverName:servers) {
            if (serverName.getServerName().equals(regionEntity.getRegionToServerName())){
                return serverName;
            }
        }
        return null;
    }

    @Override
    public int run(String[] strings) throws Exception {
        init();

        if (regionMergeEnable) {
            try {
                log.info("start region merge ...");
                autoMergeRegion();
            } catch (Exception e) {
                log.error("", e);
            }
        }

        if (regionSplitEnable) {
            try {
                log.info("start region split ...");
                autoSplitRegion();
            } catch (Exception e) {
                log.error("", e);
            }
        }

        if (tableMajorCompactEnable) {
            try {
                log.info("start table major compact ...");
                autoMajorCompact();
            } catch (Exception e) {
                log.error("", e);
            }
        }

        if (balancerEnable) {
            try {
                log.info("start balancer ...");
//                CommonTools.admin.balancer();
                autoBalance();
            } catch (Exception e) {
                log.error("", e);
            }
        }

        CommonTools.hdfs.close();

        return 0;
    }

    public static void main(String[] args) throws Exception {
        int result = ToolRunner.run(new HbaseTableAutoProcess(), args);
        System.exit(result);
    }


}
